package com.taobao.analyse.main;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextOutputFormat;

/**
 * 全量编辑记录扫描入口
 * @author xuanwei
 * @version 2011-11-27
 * @Mail-To xuanwei@taobao.com
 */
public class ArchiveMain {
	private String	baseOutPath	= "yourpath";

	public JobConf getJob(String[] args) throws Exception {
		JobConf conf = new JobConf(new Configuration(), ArchiveMain.class);
		// MISC
		conf.setJobName("DupPunish");
		conf.setMaxMapAttempts(2);
		conf.setNumReduceTasks(20);

		// MAPPER
		conf.setMapOutputKeyClass(LongWritable.class);
		conf.setMapOutputValueClass(Text.class);

		// REDUCER
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(Text.class);
		//
		//conf.setReducerClass(LCReducer.class);

		List<String> paths = new ArrayList<String>();
		paths.add("/group/tbauction/lifecycle/archive/20120111-02");
		
		//
		for (String x : paths) {
			System.out.println("SSTFile-Input:" + x);
			//
			//MultipleInputs.addInputPath(conf, new Path(x), SSTFileInputFormat.class, LCMapper.class);
		}

		// OUTPUT
		Path out = this.buildOutputPath();
		conf.setOutputFormat(TextOutputFormat.class);

		System.out.println("path:" + out.toString());
		TextOutputFormat.setOutputPath(conf, out);

		return conf;
	}

	private Path buildOutputPath() {
		SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd");
		String today = dateFormat.format(new Date());

		return new Path(baseOutPath + "pt=" + today + "000000/");
	}

	public static void main(String[] args) throws Exception {
		try {
			JobClient.runJob(new ArchiveMain().getJob(args));
		} catch (Exception e) {
			throw e;
		}
	}
}
